
                /*********************************************
                **                                          **
                **                                          **
                **                 cs.cml.C                 **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/

/*
Synchronous Communication
Check synchonous communication. 
Needs 3 processors or more to execute.
*/

#include "arch.h"

Global_id gid1;
Global_id gid2;
Coll_id cid1, cid2;

//Some parallel machines do not perform well as regards 
//to C++ static data initialization. The function below 
//forces initialization.
void init_gid(int, char **){
   gid1.set();
   gid2.set();
   cid1.set();
   cid2.set();
   fprintf(Sdf::trace, "node%d: gid1=%d, gid2=%d, cid1=%d, cid2=%d\n",
           Sdf::PROC_ID, gid1.get(), gid2.get(), cid1.get(), cid2.get());
   fflush(Sdf::trace);
}

#define B 5

ChanLD<int> *chan[4];

class Id{
   public:
        int process_Id;
        int sthread_Id;
};

ChanRD<Id> *proc_to_left;
ChanRD<Id> *proc_to_right;
int comm_rank;
int comm_size;

CommL *comm;

class Sthread: public S_Thread{
   public:
        int pid;
        int step;
        Id id;

        Sthread(int p, int s): S_Thread((Smf0_ptr)&Sthread::body){
                pid = p;
                step = s;
                id.process_Id = Sdf::PROC_ID;
                id.sthread_Id   = pid;
        }

        void body(){

                fprintf(Sdf::trace, "node%d, in sthread %d before loop at step %d\n",
                        Sdf::PROC_ID, pid, step);
                fflush(Sdf::trace);
                int i;
                for(i = 0; i < B; i++){
                   fprintf(Sdf::trace, "node%d, sthread %d in for loop, i = %d\n",
                           Sdf::PROC_ID, pid, i);
                   fflush(Sdf::trace);
                   reschedule();
                }
                fprintf (Sdf::trace, "node%d, in sthread %d after for loop at step %d\n",
                         Sdf::PROC_ID, pid, step);
                fflush(Sdf::trace);
                if(step==0 && pid==1 && comm_rank>0){
                   proc_to_left->send(id);
                   fprintf(Sdf::trace, "node%d, in sthread %d, after send left, at step %d\n",
                                Sdf::PROC_ID, pid, step);
                   fprintf(Sdf::trace, "node%d, send left rank=%d,tag_av=%d tag_ev=%d, Tsize=%d,comm=%d, rem_proc_s=%d, rem_proc_r=%d, msg_size_s=%d, msg_size_r=%d, chan_capa=%d\n",
                                Sdf::PROC_ID,
                                comm_rank,
                                proc_to_left->get_tag_av(),
                                proc_to_left->get_tag_ev(),
                                proc_to_left->get_Tsize(),
                                proc_to_left->get_comm(),
                                proc_to_left->get_remote_proc(Sdf::SND),
                                proc_to_left->get_remote_proc(Sdf::RCV),
                                proc_to_left->get_msg_size(Sdf::SND),
                                proc_to_left->get_msg_size(Sdf::RCV),
                                proc_to_left->get_chan_capacity());
                    fflush(Sdf::trace);
                }
                if(step==0 && pid==2 && comm_rank<comm_size-1){
                   proc_to_right->send(id);
                   fprintf(Sdf::trace, "node%d, in sthread %d after send right at step %d\n",
                                Sdf::PROC_ID, pid, step);
                   fprintf(Sdf::trace, "node%d, send right rank=%d,tag_av=%d tag_ev=%d, Tsize=%d,comm=%d, rem_proc_s=%d, rem_proc_r=%d, msg_size_s=%d, msg_size_r=%d, chan_capa=%d\n",
                                Sdf::PROC_ID,
                                comm_rank,
                                proc_to_right->get_tag_av(),
                                proc_to_right->get_tag_ev(),
                                proc_to_right->get_Tsize(),
                                proc_to_right->get_comm(),
                                proc_to_right->get_remote_proc(Sdf::SND),
                                proc_to_right->get_remote_proc(Sdf::RCV),
                                proc_to_right->get_msg_size(Sdf::SND),
                                proc_to_right->get_msg_size(Sdf::RCV),
                                proc_to_right->get_chan_capacity());
                    fflush(Sdf::trace);

                }
                int index;
                union S_or_T{
                   Thread  *T;
                   Sthread *S;
                } f_ptr;
                f_ptr.T = father_ptr;
                if(step==1){
                   if( (f_ptr.S)->pid == 1 )
                      index = pid%10 - 1;
                   else
                      index = pid%20 + 1;
                   chan[index]->send(pid);
                   fprintf(Sdf::trace, "node%d, sthread%d after send,step %d,index=%d,data sent:%d\n",
                           Sdf::PROC_ID, pid, step, index, pid);
                   fflush(Sdf::trace);

                }
                if(step<1){
                   Sthread *sthread1 = new Sthread(pid*10 + 1, step+1);
                   Sthread *sthread2 = new Sthread (pid*10 + 2, step+1);
                   S_Thread *st_table[2] = {sthread1, sthread2};
                   int pri[2] = {1, 1};
                   par(2, st_table, pri);
                   delete sthread1, delete sthread2;
                }
                fprintf(Sdf::trace, "node%d, sthread %d the end at step %d\n",
                             Sdf::PROC_ID, pid, step);
                fflush(Sdf::trace);

         }
};

class Sync: public S_Thread{
   public:
        int pid[4];
        Id id_left, id_right;

        Sync()
        : S_Thread((Smf0_ptr)&Sync::body){}

        void body(){

                fprintf(Sdf::trace, "node%d, in Sync0\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
                Alts *alt_table2[2];
                if(comm_rank > 0 && comm_rank < comm_size-1){
                   alt_table2[0] =
                   new AltRDrecv<Id>(proc_to_left,
                                     this,
                                     (Smf2_ptr)&Sync::alt_body2,
                                     proc_to_left,
                                     id_left);
                   alt_table2[1] =
                   new AltRDrecv<Id>(proc_to_right,
                                     this,
                                     (Smf2_ptr)&Sync::alt_body2,
                                     proc_to_right,
                                     id_right);
                   AltCtrl *alt_ctrl2 = new AltCtrl(2, alt_table2);
                   for(int i = 0; i< 2; i++){
                      Sdf::alt(alt_ctrl2);
                   }
                   delete alt_ctrl2;
                   delete alt_table2[1], delete alt_table2[0];

                   fprintf(Sdf::trace, "node%d, sync1\n", Sdf::PROC_ID);
                   fflush(Sdf::trace);
                }
                if (comm_rank == 0){
                  proc_to_left->recv(id_left);
                  fprintf(Sdf::trace, "node%d, sync2,id_left.prcr_Id=%d,id_left.thdr_Id=%d\n",
                               Sdf::PROC_ID, id_left.process_Id, id_left.sthread_Id);
                  fprintf(Sdf::trace, "node%d, recv left rank=%d,tag_av=%d tag_ev=%d,Tsize=%d,comm=%d,rem_proc_s=%d, rem_proc_r=%d\n",
                                Sdf::PROC_ID,
                                comm_rank,
                                proc_to_left->get_tag_av(),
                                proc_to_left->get_tag_ev(),
                                proc_to_left->get_Tsize(),
                                proc_to_left->get_comm(),
                                proc_to_left->get_remote_proc(Sdf::SND),
                                proc_to_left->get_remote_proc(Sdf::RCV));
                  fflush(Sdf::trace);

                }
                if (comm_rank == comm_size-1){
                   proc_to_right->recv(id_right);
                   fprintf(Sdf::trace, "node%d, sync3,id_right.prcr_Id=%d,id_right.thdr_Id=%d\n",
                               Sdf::PROC_ID, id_right.process_Id, id_right.sthread_Id);
                   fprintf(Sdf::trace, "node%d, recv right, rank=%d,tag_av=%d tag_ev=%d,Tsize=%d,comm=%d,rem_proc_s=%d, rem_proc_r=%d\n",
                                Sdf::PROC_ID,
                                comm_rank,
                                proc_to_right->get_tag_av(),
                                proc_to_right->get_tag_ev(),
                                proc_to_left->get_Tsize(),
                                proc_to_right->get_comm(),
                                proc_to_right->get_remote_proc(Sdf::SND),
                                proc_to_right->get_remote_proc(Sdf::RCV));
                   fflush(Sdf::trace);

                }
                Alts *alt_table4[4];
                alt_table4[0] =
                new AltLDrecv<int>(chan[0],
                                   this,
                                   (Smf2_ptr)&Sync::alt_body4,
                                   0,
                                   pid[0]);
                alt_table4[1] =
                new AltLDrecv<int>(chan[1],
                                   this,
                                   (Smf2_ptr)&Sync::alt_body4,
                                   0,
                                   pid[1]);
                alt_table4[2] =
                new AltLDrecv<int>(chan[2],
                                   this,
                                   (Smf2_ptr)&Sync::alt_body4,
                                   0,
                                   pid[2]);
                alt_table4[3] =
                new AltLDrecv<int>(chan[3],
                                   this,
                                   (Smf2_ptr)&Sync::alt_body4,
                                   0,
                                   pid[3]);

                AltCtrl *alt_ctrl4 = new AltCtrl(4, alt_table4);
                for(int i = 0; i< 4; i++){
                   Sdf::alt(alt_ctrl4);
                }
                fprintf(Sdf::trace, "node%d, sync3\n", Sdf::PROC_ID);

                delete alt_ctrl4; 
                delete alt_table4[3]; delete alt_table4[2];
                delete alt_table4[1]; delete alt_table4[0];

                fprintf(Sdf::trace, "node%d, sync4\n", Sdf::PROC_ID);
                fflush(Sdf::trace);

        }

        void alt_body2(int alt_rk, void *arg_ptr){
             //notice the first implicit parameter
            if(alt_rk==0)
               fprintf(Sdf::trace, "node%d, alt_body2.%d,id_left.prcr_Id=%d,id_left.thdr_Id=%d\n",
                           Sdf::PROC_ID, alt_rk, id_left.process_Id, id_left.sthread_Id);
            if(alt_rk==1)
               fprintf(Sdf::trace, "node%d, alt_body2.%d,id_right.prcr_Id=%d,id_right.thdr_Id=%d\n",
                           Sdf::PROC_ID, alt_rk, id_right.process_Id, id_right.sthread_Id);
            ChanRD<Id> *chan_ptr = (ChanRD<Id> *)arg_ptr;
            fprintf(Sdf::trace, "node%d, alt_body2, rank=%d,tag_av=%d tag_ev=%d,Tsize=%d,comm=%d,rem_proc_s=%d, rem_proc_r=%d\n",
                                Sdf::PROC_ID,
                                comm_rank,
                                chan_ptr->get_tag_av(),
                                chan_ptr->get_tag_ev(),
                                proc_to_left->get_Tsize(),
                                chan_ptr->get_comm(),
                                chan_ptr->get_remote_proc(Sdf::SND),
                                chan_ptr->get_remote_proc(Sdf::RCV));
            fflush(Sdf::trace);

        }

        void alt_body4(int alt_rk, void *arg_ptr){
            //notice the first implicit parameter
            fprintf(Sdf::trace, "node%d, alt_proc%d, pid[%d] = %d, &pid[%d] = %d\n",
                    Sdf::PROC_ID, 
                    alt_rk, alt_rk, pid[alt_rk], alt_rk, &pid[alt_rk]);
            fflush(Sdf::trace);
        }
};

class Root_thread: public Thread{
   public:
        Root_thread()
        : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){

               fprintf(Sdf::trace, "node%d, in Root_thread::body0\n", Sdf::PROC_ID);
               fflush(Sdf::trace);

               int colour;
               if(Sdf::PROC_ID==Sdf::PROC_NB-1)
                  colour=Sdf::UNDEFINED;
               else
                  colour=1;

                 //UNDEFINED is not counted as a colour
               CommL **comm_table = Sdf::COMML_WORLD->split(cid1, 1, colour, Sdf::PROC_NB-2-Sdf::PROC_ID);
               comm = *comm_table;

               fprintf(Sdf::trace, "node%d, colour=%d, comm=%d\n",
                       Sdf::PROC_ID, colour, comm);
               fflush(Sdf::trace);

                 //last process is just terminated
               if(Sdf::PROC_ID==Sdf::PROC_NB-1)
                  return;

               comm_rank = comm->rank();
               comm_size = comm->size();
               fprintf(Sdf::trace, "node%d, in Root_thread::body1, comm_rank=%d, comm_size=%d\n", 
                       Sdf::PROC_ID,comm_rank, comm_size);
               fflush(Sdf::trace);

               proc_to_left = new ChanRD<Id>();
                             //synchronous
               proc_to_right = new ChanRD<Id>();
                             //synchronous
               int left_id = ((comm_rank > 0)? comm_rank-1 : Sdf::NOPID);
               proc_to_left->set(gid1, left_id, 1, comm);

               int right_id = ((comm_rank < comm_size-1)? comm_rank+1 : Sdf::NOPID);
               proc_to_right->set(gid2, right_id, 1, comm);

               Sdf::barrier(cid2, comm);
                //synchronizes channel construction over comm (no longer
                //necessary starting from version 5)

               chan[0]      = new ChanLD<int>;
               chan[1]      = new ChanLD<int>;
               chan[2]      = new ChanLD<int>;
               chan[3]      = new ChanLD<int>;
               Sthread *sthread1= new Sthread (1, 0);
               Sthread *sthread2= new Sthread (2, 0);
               Sync *sync   = new Sync;

               S_Thread *st_table[3] = {sthread1, sthread2, sync};
               int pri[3] = {1, 1, 1};
               par(3, st_table, pri);

               fprintf(Sdf::trace, "node%d, in Root_thread::body2\n", Sdf::PROC_ID);
               fflush(Sdf::trace);

               Sdf::barrier(cid2, comm);
                     //synchronizes chan destruction
               delete proc_to_left, delete proc_to_right;
               delete chan[0],  delete chan[1], 
               delete chan[2],  delete chan[3];
               delete sthread1, delete sthread2;
               delete comm;
       }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)
